//package com.navinfo.opentsp.platform.computing.analysis.application;
//
//import com.mongodb.spark.MongoSpark;
//import com.mongodb.spark.config.ReadConfig;
//import com.mongodb.spark.rdd.api.java.JavaMongoRDD;
//import com.navinfo.opentsp.platform.computing.analysis.util.PropertiesUtil;
//import com.navinfo.stream.qingqi.protocol.java.LCLocationData;
//import org.apache.kafka.clients.producer.ProducerConfig;
//import org.apache.spark.api.java.JavaSparkContext;
//import org.apache.spark.api.java.function.Function2;
//import org.apache.spark.api.java.function.PairFlatMapFunction;
//import org.apache.spark.api.java.function.VoidFunction;
//import org.apache.spark.sql.SparkSession;
//import org.bson.Document;
//import org.bson.types.Binary;
//import org.springframework.kafka.core.DefaultKafkaProducerFactory;
//import org.springframework.kafka.core.KafkaTemplate;
//import scala.Tuple2;
//
//import java.util.*;
//
//public class MongoToKafka {
//
//    public static void main(String[] args) {
//        String day = args[0];
//        String mongoUri = PropertiesUtil.getProperties("mongoUri");
////        if(args.length >= 2){
////            day = args[0];
////            mongoUri = args[1];
////        }
//        SparkSession spark = SparkSession.builder()
//                .appName("MongoToKafka")
////                .master("local")
//                .config("spark.mongodb.input.uri", mongoUri)
//                .config("spark.mongodb.input.partitioner", "MongoPaginateBySizePartitioner")
//                .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
//                .getOrCreate();
//        try{
//            JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
//            exec(day, jsc);
//        } catch (Exception e) {
//            System.out.println(e.getMessage());
//        } finally {
//            spark.stop();
//        }
//    }
//
//    private static void exec(String day, JavaSparkContext jsc) {
//        Map<String, String> readOverrides = new HashMap<>();
//        String topic = PropertiesUtil.getProperties("topic");
//        String kafka = PropertiesUtil.getProperties("kafka");
//        readOverrides.put("collection", "GpsDataEntity_"+day.substring(2,6));
////        readOverrides.put("readPreference.name", "secondary");
//        ReadConfig readConfig = ReadConfig.create(jsc).withOptions(readOverrides);
//        Document filter = Document.parse("{ $match: {day : '"+ day + "'} }");
//        JavaMongoRDD<Document> rdd = MongoSpark.load(jsc, readConfig).withPipeline(Collections.singletonList(filter));
//        System.out.println(rdd.count() + "===================================");
//        rdd.mapPartitionsToPair((PairFlatMapFunction<Iterator<Document>, Long, List<Document>>) itor -> {
//            List<Tuple2<Long, List<Document>>> ret = new ArrayList<>();
//            while(itor.hasNext()){
//                Document doc = itor.next();
//                Long tid = doc.getLong("tId");
//                List<Document> dataList = (List<Document>) doc.get("dataList");
//                ret.add(Tuple2.apply(tid, dataList));
//            }
//            return ret.iterator();
//        }).reduceByKey((Function2<List<Document>, List<Document>, List<Document>>) (elements1, elements2) -> {
//            elements1.addAll(elements2);
//            return elements1;
//        }).foreachPartition(new VoidFunction<Iterator<Tuple2<Long, List<Document>>>>() {
//            @Override
//            public void call(Iterator<Tuple2<Long, List<Document>>> itor) throws Exception {
//                Map<String, Object> props = new HashMap<>();
//                props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
////        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.30.50.21:9092");
//                props.put(ProducerConfig.RETRIES_CONFIG, 0);
//                props.put(ProducerConfig.BATCH_SIZE_CONFIG, 100000);
//                props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
//                props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
//                props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//                props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "com.navinfo.opentsp.platform.computing.analysis.entity.ProtoBufSerializer");
//                props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
//
//                KafkaTemplate<String, Object> kafkaTemplate = new KafkaTemplate<String, Object>(new DefaultKafkaProducerFactory<>(props));
//                while(itor.hasNext()){
//                    Tuple2<Long, List<Document>> tuple2 = itor.next();
//                    long tid = tuple2._1;
//                    List<Document> dataList = tuple2._2;
//                    for (Document data : dataList) {
//                        Binary binary = (Binary) data.get("data");
//                        LCLocationData.LocationData locationData = null;
//                        try {
//                            locationData = LCLocationData.LocationData.parseFrom(binary.getData());
//                        } catch (Exception e) {
//                            //                        logger.error("", e);
//                            continue;
//                        }
//                        LCLocationData.LocationData newLocationData = locationData.toBuilder()
////                                .setGpsDate(System.currentTimeMillis()/1000)
//                                .setTerminalId(tid)
//                                .build();
////                    Packet outPacket = new Packet();
////                    outPacket.setCommand(12290);
////                    outPacket.setProtocol(2);
////                    outPacket.setUniqueMark(tid + "");
////                    outPacket.setSerialNumber(1111);
////                    outPacket.setContent(newLocationData.toByteArray());
//
////                    KafkaCommand kafkaCommand = new KafkaCommand();
////                    kafkaCommand.setKey(tid + "");
////                    kafkaCommand.setCommandId("3002");
////                    kafkaCommand.setSendTime(new Date());
////                    kafkaCommand.setSerialNumber(11111);
////                    kafkaCommand.setTopic("0200spark");
////                    kafkaCommand.setMessage(JsonUtil.pojo2Bytes(outPacket));
////                    kafkaTemplate.send("0200spark", tid + "", JsonUtil.pojo2Json(kafkaCommand));
//                        kafkaTemplate.send(topic, tid + "", newLocationData.toByteArray());
//                    }
//                }
//            }
//        });
//    }
//
//    public static Map<String, Object> producerConfigs(String kafka) {
//        Map<String, Object> props = new HashMap<>();
//        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
////        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.30.50.21:9092");
//        props.put(ProducerConfig.RETRIES_CONFIG, 0);
//        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 100000);
//        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
//        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
//        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "com.navinfo.opentsp.platform.computing.analysis.entity.ProtoBufSerializer");
//        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
//        return props;
//    }
//}
